package com.chukun.multi.masterslave;

import com.chukun.multi.twophase.AbstractTerminatableThread;

import java.io.BufferedReader;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;

/**
 * @author chukun
 * 子线程执行任务
 */
public class Worker extends AbstractTerminatableThread {

    private static final Pattern SPLIT_PATTERN = Pattern.compile("\\|");

    private final Map<String, AtomicInteger> repository;

    private final BlockingQueue<BufferedReader> workQueue;

    private final String selfDevice = "ESB";
    private final String excludedOperationNames;
    private final String includedOperationNames;
    private final String destinationSysName;

    public Worker(Map<String, AtomicInteger> repository, String excludedOperationNames,
                  String includedOperationNames, String destinationSysName) {
        this.repository = repository;
        this.workQueue = new ArrayBlockingQueue<>(100);
        this.excludedOperationNames = excludedOperationNames;
        this.includedOperationNames = includedOperationNames;
        this.destinationSysName = destinationSysName;
    }

    public void submitWorkload(BufferedReader taskWorkload) {
        try {
            workQueue.put(taskWorkload);
            terminationToken.reservations.incrementAndGet();
        } catch (InterruptedException e) {
            ;
        }
    }

    @Override
    protected void doRun() throws Exception {
        BufferedReader logFileReader = workQueue.take();
        String interfaceLogRecord;
        String[] recordParts;
        String timeStamp;
        AtomicInteger reqCount;
        AtomicInteger existingReqCounter;
        int i = 0;
        try {
            while ((interfaceLogRecord = logFileReader.readLine()) != null) {
                recordParts = SPLIT_PATTERN.split(interfaceLogRecord, 0);

                // 避免CPU占用过高,每处理100000条，让cpu休息一会，睡眠100 ms
                if (0 == (++i) % 100000) {
                    TimeUnit.MILLISECONDS.sleep(100);
                    i = 0;
                }
                // 跳过无效记录 （如果有的话）
                if (recordParts.length < 1) {
                    continue;
                }

                // 只考虑表示发送请求给selfDevice 所指定的系统的记录
                if (("request".equals(recordParts[2])) && (recordParts[6].startsWith(selfDevice))) {
                    timeStamp = recordParts[0];
                    timeStamp = new String(timeStamp.substring(0, 19).toCharArray());
                    String operName = recordParts[4];
                    reqCount = repository.get(timeStamp);
                    if (Objects.isNull(reqCount)) {
                        reqCount = new AtomicInteger(0);
                        existingReqCounter = repository.putIfAbsent(timeStamp, reqCount);
                        if (!Objects.isNull(existingReqCounter)) {
                            reqCount = existingReqCounter;
                        }
                    }
                    if (isSrcDeviceEEligible(recordParts[5])) {
                        if (excludedOperationNames.contains(operName + ",")) {
                            continue;
                        }
                        if ("*".equals(includedOperationNames)) {
                            reqCount.incrementAndGet();
                        } else {
                            if (includedOperationNames.contains(operName + ",")) {
                                reqCount.incrementAndGet();
                            }
                        }
                    }
                }

            }
        } finally {
            terminationToken.reservations.decrementAndGet();
            logFileReader.close();
        }
    }

    /**
     * 判断目标设备名是否在待统计之列
     *
     * @param sourceNE
     * @return
     */
    private boolean isSrcDeviceEEligible(String sourceNE) {
        boolean result = false;
        if ("*".equals(destinationSysName)) {
            result = true;
        } else if (destinationSysName.equals(sourceNE)) {
            result = true;
        }
        return result;
    }
}
